package com.atguigu.flink.datastream.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.internals.Topic;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;

/**
 * @author WEIYUNHUI
 * @date 2023/6/13 9:13
 * <p>
 * Kafka Connector :
 * 1. 从kafka中消费数据
 * 1.1 消费者对象:
 * KafkaConsumer
 * 1.2 消费者相关配置：
 * key/value的反序列化器:
 * key.deserializer => org.apache.kafka.common.serialization.StringDeserializer
 * value.deserializer => org.apache.kafka.common.serialization.StringDeserializer
 * 集群地址:
 * bootstrap.servers  => hadoop102:9092,hadoop103:9092,hadoop104:9092
 * 消费者组:
 * group.id  => 随便
 * offset自动提交:
 * enable.auto.commit  => true / false
 * offset自动提交的间隔:
 * auto.commit.interval.ms => 5000
 * offset重置：
 * auto.offset.reset => latest（尾） | earliest（头）
 * 重置的情况:
 * 1. 新的消费者组 ， 之前没有消费过数据， 也就没有对应的offset
 * 2. 旧的消费者组 ， 之前有消费过数据，但是对应的offset已经在kafka中不存在（可能的原因是Kafka默认7天会清理数据）
 * <p>
 * 1.3 Kafka提供的消费者的配置类 ConsumerConfig
 * <p>
 * 2. 往kafka生产数据
 */
public class Flink03_KafkaSource {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        /*
                封装要从哪个主题那个分区的offset进行消费
                TopicPartition topicPartition = new TopicPartition("topicA", 3);
                HashMap<TopicPartition, Long> offsets = new HashMap<>();
                offsets.put(topicPartition , 40L) ;
         */

        KafkaSource<String> kafkaSource =
                KafkaSource.<String>builder()
                        .setBootstrapServers("hadoop102:9092,hadoop103:9092")
                        .setGroupId("flink1")
                        .setTopics("topicA")
                        .setValueOnlyDeserializer(new SimpleStringSchema()) //仅针对于没有key的消息
                        //.setDeserializer() //针对于key和value
                        // 默认使用记录的offset进行消费 ，如果涉及到重置offset， 选择重置到尾
                        //.setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                        // 默认使用记录的offset进行消费 ，如果涉及到重置offset， 选择重置到头
                        .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
                        // 指定Offset消费
                        //.setStartingOffsets(OffsetsInitializer.offsets(offsets))
                        //如果其他的设置没有对应的方法， 统一使用.setProperty( 配置项 ， 配置值)
                        // .setProperty("enable.auto.commit" , "true" )
                        .setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
                        //.setProperty("auto.commit.interval.ms" , "6000")
                        .setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "6000")
                        .build();

        DataStreamSource<String> ds = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource");

        ds.print();
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
